package com.yanqu.data.analytics.consumer;


import cn.thinkingdata.tga.javasdk.inter.Consumer;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.yanqu.road.utils.ZipUtils;
import com.yanqu.road.utils.http.HttpHelper;
import org.apache.logging.log4j.Logger;

import java.util.*;

public class CollectConsumer  implements Consumer, ConsumerQueueOperate {

    Logger logger;

    private String serverUrl;

    private static final int maxSize = 50001;

    private static Queue<Map<String, Object>> messageQueue = new LinkedList<>();

    public CollectConsumer(Logger logger) {
        this.logger = logger;
    }

    @Override
    public void add(Map<String, Object> map) {
        synchronized (messageQueue){
            messageQueue.add(map);
            checkQueue();
        }
    }

    public void checkQueue() {
        int queueSize = messageQueue.size();
        if (queueSize >= maxSize) {
            List<Map<String, Object>> tempList = poll(queueSize - maxSize);
            logger.warn("CollectConsumer Queue size exceeded,over size:{}", queueSize - maxSize);
        }

    }

    public List<Map<String, Object>> poll(int max_num) {
        List<Map<String, Object>> list = new ArrayList<>();
        synchronized (messageQueue) {
            for (int i = 0; i < max_num; i++) {
                Map<String, Object> data = messageQueue.poll();
                if (data == null) {
                    break;
                }
                list.add(data);
            }
            return list;
        }
    }

    @Override
    public void flush() {
        try{
            for (int i = 0; i < 2; i++) {
                List<Map<String, Object>> tempList = poll(1000);
                if(tempList.size() > 0){
                    StringBuilder stringBuffer = new StringBuilder();
                    for (Map<String, Object> objectMap : tempList) {
                        String data = JSON.toJSONStringWithDateFormat(objectMap, "yyyy-MM-dd HH:mm:ss.SSS", new SerializerFeature[0]);
                        stringBuffer.append(data).append("\n");
                    }
                    long start = System.nanoTime();
                    String gzip = ZipUtils.gzip(stringBuffer.toString());
                    long count = (System.nanoTime() - start) / 1000000;
                    Map<String, String> result = HttpHelper.httpPostJson(serverUrl + "&gzip=1", gzip, "UTF-8", null, 5000, 5000, 5000);
                    String code = result.getOrDefault("code", "0");
                    if(!"200".equals(code)){
                        //重新塞回队列
                        synchronized (messageQueue){
                            messageQueue.addAll(tempList);
                        }
                        logger.error("CollectConsumer flush size {}, length {} fail, code {}", tempList.size(), gzip.length(), code);
                        //错误就不在尝试了
                        break;
                    }else {
                        logger.debug("CollectConsumer flush size {}, length {}， zip cost {}", tempList.size(), gzip.length(), count);
                    }
                }else {
                    logger.info("CollectConsumer flush size 0");
                }
            }
        }catch (Exception e){
            logger.error("CollectConsumer flush error:", e);
        }
    }

    @Override
    public void close() {

    }

    @Override
    public Queue<Map<String, Object>> getQueue() {
        return messageQueue;
    }

    @Override
    public boolean cleanQueue() {

        return true;
    }

    public String getServerUrl() {
        return serverUrl;
    }

    public void setServerUrl(String serverUrl) {
        this.serverUrl = serverUrl;
    }
}
